<?php
namespace app\command;

use app\common\model\system\SqlToRedis;
use mall\library\Huobi;
use think\console\Command;
use think\console\Input;
use think\console\Output;
use think\Exception;
use think\facade\Db;


/**
 * 获取历史火币数据，并进行保存
 * Class HistoryTrade
 * @package app\command
 */
class HistoryTrade extends Command
{
    // 记录自增表ID
    const TABLE_AUTO_ID = 'auto_id';

    // 设置表名
    const TABLE_HISTORY = 'history';

    // 设置需要查询的表
    const TABLE_SELECT_LIST = 'select_list';

    // 执行中的状态
    const TABLE_SELECT_RUN = 'select_run';

    // 存放自定义价格表
    const TABLE_INPUT_PRICE = 'input_price';

    // 耗时记录表
    const TABLE_TAKE_TIME = 'take_time_log';

    /**
     * ENV 配置中监控端口的key
     */
    const ENV_KEY_TCP_QUERY_PORT = 'history_trade.tcp_query_port';


    /**
     * websocket参数
     */
    const WEBSOCKET_PARAMS = [
        'TABLE_WEBSOCKET_TO_SYMBOL' => 'websocket_to_symbol',   //设置进程获取试试行情对应币种
        'TABLE_SYMBOL_TO_CALCULATION'   => 'symbol_to_calculation', //币种对应的计算进程
        'API'   => 'wss://api.huobi.pro/ws',  //接口地址
        'API_AWS'   => 'wss://api-aws.huobi.pro/ws',  // aws api接口，提高效率
        'PROCESS_NUM'   => 8,   //读取行情进程数
        'RESTART_TIME'  => 8,   //获取等待时间，单位 s 【未获取到数据重链】
        'SYMBOL_UPDATE_TIME'    => 2,   //更新时间
    ];

    /**
     * 获取用户规则配置
     */
    const USER_RULE_CONFIG = [
        'TABLE_NAME'    => 'user_rule_config',
        'INTERVAL_TIME' => 15,   //间隔刷新时间，单位s
        'TYPE_LIST' => [    // 类型
            'HISTORY'   => 'history',
            'RULE_CALCULATION'  => 'culation',
        ],
    ];




    /**
     * while循环的间隔时间
     * @var int
     */
    private $usleepTimeDefault = 100000;


    const TAKE_TIME_TYPE_LIST = [
        'READ_BEFOR'    => 1,   //读取数据之前
        'READ_END'    => 2,   //读取第三方结束
        'READ_INSERT'     => 3,   //读取后插入数据
        'DISTRIBUTION'  => 4,   //分配到各个表中
        'CALCULATION_BEFOR' => 5,   //计算之前，即计算获取数据
        'CALCULATION'   => 6,   //计算
        'CALL_APPLY'    => 7,   //请求第三方接口申请（写table表中）
        'CALL_RUN'      => 8,   //执行请求第三方接口
        'CALL_END'      => 9,   //结束请求第三方接口
        'RESULT_RUN'    => 10,   //查询结果
        'RESULT_END'    => 11,  //完成结果查询
        'COUNT'    => 20,  //统计用时
        'ERROR' => 21,  //用于记录错误
    ];

    /**
     * ENV 配置参数KEY列表
     */
    const ENV_KEY_LIST = [
        'IS_AWS'    => 'huobi.in_aws',  //判断是否为亚马逊服务器

        'INPUT_PRICE'   => 'call_api.is_input_price',
        'TAKE_TIME'     => 'monitor.open_take_time',
        'TAKE_TIME_RECORD_TYPE_LIST'    => 'monitor.type_list',
        'TAKE_TIME_RECORD_SYMBOL_LIST'    => 'monitor.symbol_list',

        'GET_TRADE_TYPE'    => 'call_api.get_trade_type',
    ];

    const TCP_WORKER_NUM = 4; //tcp执行进程数

    /**
     * 自定义价格
     * @var null
     */
    private $isInputPrice = null;

    /**
     * 判断是否开启耗时统计
     * @var null
     */
    static $isOpenTakeTime = null;

    static $typeListTakeTime = null;

    static $symbolListTakeTime = null;

    static $portIndex = null;




    /**
     * 重写configure
     * {@inheritdoc}
     */
    protected function configure()
    {
        // 命令的名字（"think" 后面的部分）
        $this->setName('HistoryTrade')
            // 配置一个参数 使用$input->getArgument('username')获取
            ->addArgument('portIndex')  // 端口索引
            ->addArgument('type')  // 判断操作类型
            // 运行 "php think list" 时的简短描述
            ->setDescription('实时获取货币数据，并进行保存处理。')
            // 运行命令时使用 "--help" 选项时的完整命令描述
            ->setHelp("一直执行 无参数");
    }



    /**
     *  * 重写execute
     *  * {@inheritdoc}
     *
     * @param Input $input
     * @param Output $output
     */
    protected function execute(Input $input, Output $output)
    {
        // 设置端口开始索引【增加支持多开处理】
        static::$portIndex = intval($input->getArgument('portIndex'));
        switch(strtolower($input->getArgument('type'))) {
            case 'checkuprestart': // 检查状态
                return $output->write(RestartDetection::HistoryTrade(static::$portIndex));
        }

        // 判断是否为手动定义价格
        $this->isInputPrice = env('call_api.is_input_price');
        static::$isOpenTakeTime = env(static::ENV_KEY_LIST['TAKE_TIME']);
        static::$typeListTakeTime =  array_filter(explode(',',env(static::ENV_KEY_LIST['TAKE_TIME_RECORD_TYPE_LIST'])));
        static::$symbolListTakeTime =  array_filter(explode(',',env(static::ENV_KEY_LIST['TAKE_TIME_RECORD_SYMBOL_LIST'])));

        // 创建内存表
        $this->createSwooleTable();


        // 设置监控配置
        $this->setConfig();

        // 执行计算处理【获取监听数据列表】
        $calculationPidList = (new RuleCalculation(static::TABLE_HISTORY,static::TABLE_SELECT_LIST))->execute();



        // 开启监控获取行情数据
        $this->getTickersMaster($calculationPidList);

        // 创建tcp服务
        $this->createTcpQuery();

        //将日志写入文本【@2021-05-16 改为文本日志】
    //    $this->takeTimeReadFile();

        //开启等待
        Swoole::wait();


    }

    /**
     * 将耗时日志写入文本记录，并删除历史
     * @return bool
     */
    private function takeTimeReadFile() {
        if(!static::$isOpenTakeTime) {
            return true;
        }
        Swoole::startProcess(function(){
            $logTable = Swoole::tableGet(static::TABLE_TAKE_TIME);
            while(true) {
                //写入异常日志
                $path = dirname(__FILE__) . '/takeTime/';
                is_dir($path) || mkdir($path,0777,true);
                $filename = $path . 'run.log';
                $logFilename = $path . date('Ymd-H_i_s') . '.log';
                $log = '';
                foreach($logTable as $key=>$value) {
                    try {
                        $log .= json_encode($value) . "\n";
                        $logTable->del($key);
                    } catch (\Exception $e) {
                    }
                }
                if($log) {
                    file_put_contents($filename,$log ,FILE_APPEND);
                    clearstatcache();
                    if(filesize($filename) > 1024 * 1024 * 20) {
                        @copy($filename,$logFilename);
                        @unlink($filename);
                    }

                    // 查询目录下的所有文件统计大小，并设置最大值
                    try {
                        $filenameList = [];
                        $dh = opendir($path);
                        while ($file = readdir($dh)){
                            if(!preg_match('/[0-9]{8}\-[0-9]{2}\_[0-9]{2}\_[0-9]{2}\.log/',$file)) {
                                continue;
                            }
                            $filenameList[] = $file;
                        }
                        if($filenameList) {
                            rsort($filenameList);
                            $count = 0;
                            foreach($filenameList as $file) {
                                $filename = $path . $file;
                                if($count > 1024 * 1024 * 1024) {
                                    @unlink($filename);
                                }
                                $count += filesize($filename);
                            }
                        }
                    } catch (\Exception $e) {

                    }

                }

                usleep(10000);
            }
        });
    }

    /**
     * 添加日志
     * @param $params
     */
    static public function takeTime($type = 0,$symbol = '',$ts = 0,$tradeId = 0,$ruleId = 0,$runId = 0,$trade = 0){
        if($type == self::TAKE_TIME_TYPE_LIST['READ_INSERT']) { // 获取到推送数据，添加日志
           try {
               $table = Swoole::tableGet(static::TABLE_SELECT_RUN);
               $record = [
                   'lastRunId'  => $tradeId,
                   'lastScessTime'  => $ts,
                   'lastTime'   => time()
               ];
               $table->set($symbol,$record);
           } catch (\Exception $e) {

           }
        }

        if(is_null(static::$isOpenTakeTime)) {  // 未设置限制时自主设置限制
            static::$isOpenTakeTime = env(static::ENV_KEY_LIST['TAKE_TIME']);
            static::$typeListTakeTime =  array_filter(explode(',',env(static::ENV_KEY_LIST['TAKE_TIME_RECORD_TYPE_LIST'])));
            static::$symbolListTakeTime =  array_filter(explode(',',env(static::ENV_KEY_LIST['TAKE_TIME_RECORD_SYMBOL_LIST'])));
        }


        if(!static::$isOpenTakeTime || (static::$typeListTakeTime && !in_array($type,static::$typeListTakeTime)) ||
            (!empty($symbol) && static::$symbolListTakeTime && !in_array($symbol,static::$symbolListTakeTime))) {
            return true;
        }

        if(!empty($symbol) && !is_string($symbol)) {    // 判断为异常抛出，获取错误信息
            $symbol = substr(implode('|',[$symbol->getMessage(),$symbol->getLine(),$symbol->getFile()]),0,255);
        }
        try {
            // 添加日志
            Log::write([
                'pid'   => Swoole::pid(),
                'time'  => microtime(true),
                'symbol'    => $symbol,
                'ts'    => $ts,
                'trade-id'  => $tradeId,
                'rule_id'   => $ruleId,
                'run_id'   => $runId,
                'trade_id'   => $trade,
                'type'  => $type,
            ],Log::LOG_TYPE_LIST['TAKE_TIME']);

            return true;

            $table = Swoole::tableGet(static::TABLE_TAKE_TIME);
            $table->set(static::getAutoId(static::TABLE_TAKE_TIME),[
                'pid'   => Swoole::pid(),
                'time'  => microtime(true),
                'symbol'    => $symbol,
                'ts'    => $ts,
                'trade-id'  => $tradeId,
                'rule_id'   => $ruleId,
                'run_id'   => $runId,
                'trade_id'   => $trade,
                'type'  => $type,
            ]);
        }catch (\Exception $e) {

        }
        return true;
    }


    /**
     * 查询数据
     * @param string $tablename 表明
     * @param string|null $key 键
     * @param string|null $subkey 键数组中的键
     * @return mixed
     * @throws Exception
     */
    static function tcpTableQuery(string $params,$portIndex = null) {
        $params = explode(' ',trim($params));
        $client = new \swoole_client(SWOOLE_SOCK_TCP);
        if (!$client->connect('127.0.0.1', static::tcpQueryPort($portIndex),3)) {
            throw new Exception("connect failed. Error: {$client->errCode}");
        }

        //数据分隔符
        $end = "\r\n";
        $client->send(implode(' ',$params) . $end);
        $result = json_decode($client->recv(),true);
        $client->close();
        return $result;
    }


    /**
     * 获取tcp的监控端口
     */
    static function tcpQueryPort($portIndex = null)
    {
        if(!$port = env(static::ENV_KEY_TCP_QUERY_PORT)) {
            $port = 9501;
        }
        // 获取端口
        $port += intval(is_null($portIndex)?static::$portIndex:$portIndex) * 100;
        return $port;
    }



    /**
     * 获取自增ID
     * @param string $tableName 使用到的表名
     * @return mixed 返回自增ID
     */
    static public function getAutoId(string $tableName = 'auto') {
        return Swoole::table(static::TABLE_AUTO_ID)->lockFunc(function($table) use($tableName){
            if(!$details = $table->get($tableName)) {
                $details = [
                    'id'    => 0,
                ];
            }
            $details['id']++;
            $table->set($tableName,$details);
            return $details['id'];
        });
    }

    /**
     * 创建进程所需要的内存表（仅线程内操作）
     */
    private function createSwooleTable()
    {
        $sizeUnit = 1024;

        /************* 记录最后刷新用户配置时间表 **************/
        // 写入状态表
        Swoole::table(self::USER_RULE_CONFIG['TABLE_NAME'],[
            ['last_time','TYPE_INT',11],
        ],$sizeUnit,true);
        /*********** END 记录最后刷新用户配置时间表 ************/

        /************* 记录自增表ID **************/
        // 写入状态表
        Swoole::table(static::TABLE_AUTO_ID,[
            ['id','TYPE_INT',4],
        ],$sizeUnit,true);
        /*********** END 记录自增表ID ************/

        /******* socket订阅表 ********/
        Swoole::table(static::WEBSOCKET_PARAMS['TABLE_WEBSOCKET_TO_SYMBOL'],[
            ['symbol_list','TYPE_STRING', 1024 * 10],    // 获取ID
        ],$sizeUnit);
        /***** END socket订阅表 ******/

        /******* 币种对应计算ID ********/
        Swoole::table(static::WEBSOCKET_PARAMS['TABLE_SYMBOL_TO_CALCULATION'],[
            ['calculation_id','TYPE_INT',4],    //计算ID
        ],$sizeUnit * 10);
        /***** END 币种对应计算ID ******/


        /************* 关注货币表 **************/
        Swoole::table(static::TABLE_SELECT_LIST,[
            ['ruleIdList','TYPE_STRING',1024],    //火币关注表
        ],$sizeUnit);
        /*********** END 关注货币表 ************/

        /************* 关注货币表（执行中状态） **************/
        Swoole::table(static::TABLE_SELECT_RUN,[
            ['lastRunId','TYPE_STRING',20],  //最后一次执行的识别ID
            ['lastScessTime','TYPE_STRING',15],    //最后一次执行成功时间
            ['lastTime','TYPE_STRING',15],    //最后一次查询时间
        ],$sizeUnit,true);
        /*********** END 关注货币表（执行中状态） ************/

        /*********** 判断是否保存自定义价格表 ***********/
        if($this->isInputPrice) {
            Swoole::table(static::TABLE_INPUT_PRICE,[
                ['price','TYPE_STRING',10000],    //保存价格列表
            ],$sizeUnit,true);
        }
        /********* END 判断是否保存自定义价格表 *********/

        /*********** 判断是否开启耗时统计 ***********/
        if(static::$isOpenTakeTime) {
            Swoole::table(static::TABLE_TAKE_TIME,[
                ['pid','TYPE_INT',10],  // 记录数据的PID
                ['time','TYPE_STRING',15],  //操作时间
                ['symbol','TYPE_STRING',255],  //交易币种
                ['ts','TYPE_STRING',15],    // 交易时间，毫秒
                ['trade-id','TYPE_STRING',15],  //交易唯一ID
                ['rule_id','TYPE_INT',10],  //计算中的规则ID
                ['run_id','TYPE_INT',10],  //执行计算的ID
                ['trade_id','TYPE_INT',10],  //买卖记录ID
                ['type','TYPE_INT',2],  //记录类型
            ],$sizeUnit * 100);
        }
        /********* END 判断是否开启耗时统计 *********/



    }



    /**
     * 设置监控配置
     */
    private function setConfig()
    {
        // 监控表
        $selectListTable = Swoole::table(static::TABLE_SELECT_LIST);

        for($i=0;$i<1;++$i) {    //设置开启线程数
            Swoole::startProcess(function() use($selectListTable){
                $userSymbolLastMd5  = '';
                while (true) {
                    // 获取随时变化配置信息
                    try {   //避免redis读取错误
                        list($strategyList) = $this->getUserKeysAndStrategys();
                    } catch (\Exception $e) {
                        usleep($this->usleepTimeDefault);
                        continue;
                    }


                    /************* 更新监控表 ****************/
                    $userSymbolNewMd5 = md5(json_encode($strategyList));
                    if($userSymbolNewMd5 != $userSymbolLastMd5) {
                        foreach($strategyList as $symbol=>$value) {
                            $selectListTable->set($symbol,$value);
                        }
                        foreach($selectListTable as $symbol=>$value) {
                            if(!array_key_exists($symbol,$strategyList)) {
                                $selectListTable->del($symbol); //删除监控
                            }
                        }
                    }
                    $userSymbolLastMd5 = $userSymbolNewMd5;
                    /*********** END 更新监控表 **************/
                    usleep($this->usleepTimeDefault);
                }
            });
        }
    }


    /**
     * 获取并将数据写入
     * @param $index
     * @param $calculationProcessList
     */
    private function getTickesFromWebSocket($websocketIndex,$calculationProcessList) {
        // 获取行情方式类型 0，1 为队列；2为请求api
        $getTradeType = intval(env(self::ENV_KEY_LIST['GET_TRADE_TYPE'],0));
        $getTradeType = in_array($getTradeType,[1,2,3])?$getTradeType:1;
        static::takeTime(static::TAKE_TIME_TYPE_LIST['ERROR'],"start(" . Swoole::pid() . ")");
        $calculationTable = Swoole::tableGet(static::WEBSOCKET_PARAMS['TABLE_SYMBOL_TO_CALCULATION']);
        $websocketSymbolTable = Swoole::tableGet(static::WEBSOCKET_PARAMS['TABLE_WEBSOCKET_TO_SYMBOL']);

        switch($getTradeType) {
            case 1: // 推送方式
                $client = new \WebSocket\Client(static::WEBSOCKET_PARAMS[env(self::ENV_KEY_LIST['IS_AWS'])?'API_AWS':'API'],[
                    'filter'    => ['binary'],
                    'timeout'   => static::WEBSOCKET_PARAMS['RESTART_TIME'],
                ]);
                break;
            case 2: // 直接调用火币API接口
                $client = new Huobi('','','');
                break;
        }

        // 币种对应计算的索引
        $symbolToCalculationIndex = [];
        $symbolLastGetTime = [];    //已监听的币种
        $symbolList = [];   //监听的币种
        $lastCountTime = 0;
        while (true) {
            try {
                if(time() - $lastCountTime > static::WEBSOCKET_PARAMS['SYMBOL_UPDATE_TIME']) {
                    /************** 获取监听的币种，并进行监听及取消 *************/

                    $symbolList = $websocketSymbolTable->get($websocketIndex);
                    $symbolList = is_array($symbolList)?explode(',',$symbolList['symbol_list']):[];

                    if(!is_array($symbolList)) {
                        throw new \Exception('not find');
                    }

                    $symbolList = array_filter($symbolList);
                    foreach($symbolList as $symbol) {
                        if($getTradeType == 1 && (!array_key_exists($symbol,$symbolLastGetTime) || time() - $symbolLastGetTime[$symbol] > static::WEBSOCKET_PARAMS['RESTART_TIME'])) {
                            // 订阅接收信息
                            $client->text(json_encode([
                                'sub'	=> 'market.' . $symbol . '.trade.detail',
                                'id'	=> 'id' . $symbol
                            ]));
                        }
                    }
                    foreach($symbolLastGetTime as $symbol=>$del) {
                        if($getTradeType == 1 && !in_array($symbol,$symbolList)) {
                            // 取消订阅
                            $client->text(json_encode([
                                'unsub'	=> 'market.' . $symbol . '.trade.detail',
                                'id'	=> 'id' . $symbol
                            ]));
                            unset($symbolLastGetTime[$symbol]);
                        }
                    }
                    /************ END 获取监听的币种，并进行监听及取消 ***********/


                    /********** 获取币种对应的计算进程 ************/
                    $symbolToCalculationIndex = [];
                    foreach($calculationTable as $symbol=>$index) {
                        if(!in_array($symbol,$symbolList)) {
                            continue;
                        }
                        $symbolToCalculationIndex[$symbol] = $index['calculation_id'];
                    }
                    /******** END 获取币种对应的计算进程 **********/

                    // 当前计算时间
                    $lastCountTime = time();
                }


                switch($getTradeType) {
                    case 1: // 推送获取数据
                        $message = $client->receive();
                        break;
                    case 2: // 请求api获取数据
                        usleep($this->usleepTimeDefault * 10);
                        if(empty($symbolList)) {
                            $message = null;
                        } else {
                            $symbol = $symbolList[array_rand($symbolList)];
                            $message = $client->get_history_trade($symbol,1);

                            $message = [
                                'ch'    => $message['ch'],
                                'tick'  => $message['data'][0],
                            ];
                            foreach($message['tick']['data'] as &$value) {
                                $value['tradeId'] = $value['trade-id'];
                            }
                            // 将数据格式化
                            $message = gzencode(json_encode($message));
                        }
                        break;
                    case 3: // 手动输入【暂未处理】
                        $message = null;
                        break;
                }

                if(empty($message) || !($message = gzdecode($message))) {
                    continue;
                }
                if(!$message = json_decode($message,true)) {
                    continue;
                }
                if(array_key_exists('ping',$message) || array_key_exists('pong',$message)) {
                    $client->text(json_encode($message));
                    continue;
                }
                if(!array_key_exists('ch',$message) || !array_key_exists('tick',$message) || !array_key_exists('data',$message['tick'])) {
                  /*  print_r($message);*/
                    continue;
                }

                // 获取到币种信息，对币种信息进行计算保存
                $symbol = explode('.',$message['ch'])[1];
                if(!in_array($symbol,$symbolList) || !array_key_exists($symbol,$symbolToCalculationIndex)) {
                   /* if($websocketIndex ==1) {
                        echo $symbol;
                        print_r($symbolList);
                        print_r($symbolToCalculationIndex);
                    }*/
                    continue;
                }
                // 保存最后获取时间
                $symbolLastGetTime[$symbol] = time();
                /********************* 送入计算进程 ***********************/
                foreach($message['tick']['data'] as $details) {
                    try{
                        // 写入计算
                        $writeData = [
                            'symbol'    => $symbol,
                            'ts'     => $details['ts'],
                            'trade-id'     => $details['tradeId'],
                            'amount'     => $details['amount'],
                            'price'     => $details['price'],
                            'direction'     => $details['direction'],
                        ];
                        $calculationProcessList[$symbolToCalculationIndex[$symbol]]->write(json_encode($writeData));
                        static::takeTime(static::TAKE_TIME_TYPE_LIST['READ_INSERT'],$symbol,$writeData['ts'],$writeData['trade-id']);
                    }catch (\Exception $e) {
                        static::takeTime(static::TAKE_TIME_TYPE_LIST['ERROR'],$e);
                    }
                }
                /******************* END 送入计算进程 *********************/


            } catch (\Exception $e) {
                if(empty($symbolList)) {
                    continue;
                }
                static::takeTime(static::TAKE_TIME_TYPE_LIST['ERROR'],$e);
                try {
                    if($getTradeType == 1) {
                        $client->disconnect();
                        $client->close();
                    }
                } catch (\Exception $e) {
                    static::takeTime(static::TAKE_TIME_TYPE_LIST['ERROR'],$e);
                }
                throw $e;
            }
        }
    }

    /**
     * 开启监控获取行情数据
     */
    private function getTickersMaster($calculationProcessList)
    {
        //开启获取行情进程
        for($i=0;$i<static::WEBSOCKET_PARAMS['PROCESS_NUM'];++$i) {
            Swoole::startProcess(function() use($i,$calculationProcessList){
                while(true) {
                    try {
                        $this->getTickesFromWebSocket($i,$calculationProcessList);
                    }catch (\Exception $e) {
                        static::takeTime(static::TAKE_TIME_TYPE_LIST['ERROR'],$e);
                    }
                    usleep($this->usleepTimeDefault);
                }
            },true);
        }


        /************* 计算监控数据对应 ****************/
        $calculationProcessNum = count($calculationProcessList);
        Swoole::startProcess(function($process) use($calculationProcessNum){
            $selectListTable = Swoole::tableGet(static::TABLE_SELECT_LIST);
            $calculationCount = [];
            for($i=0;$i<$calculationProcessNum;++$i) {
                $calculationCount[$i] = 0;
            }
            $calculationTable = Swoole::tableGet(static::WEBSOCKET_PARAMS['TABLE_SYMBOL_TO_CALCULATION']);
            $calculationList = [];

            $symbolIniList = [];
            $symbolWebsocketCount = [];
            for($i=0;$i<static::WEBSOCKET_PARAMS['PROCESS_NUM'];++$i) {
                $symbolWebsocketCount[$i] = 0;
                $symbolIniList[$i] = [];
            }
            $symbolWebsocketList = [];

            $WebsocketToSymbolTable = Swoole::tableGet(static::WEBSOCKET_PARAMS['TABLE_WEBSOCKET_TO_SYMBOL']);


            while (true) {  //循环计算
                try {
                    $symbolList = [];
                    foreach($selectListTable as $symbol=>$value) {
                        $symbolList[] = $symbol;
                        /************ 设置币种对应的计算进程 **************/
                        if(!array_key_exists($symbol,$calculationList)) {
                            $tmp = array_count_values($calculationList);
                            $sortList = $calculationCount;
                            foreach($tmp as $key=>$val) {
                                $sortList[$key] = $val;
                            }
                           asort($sortList);
                           foreach($sortList as $key=>$del) {
                               $calculationTable->set($symbol,['calculation_id'=>$key]);
                               $calculationList[$symbol] = $key;
                               break;
                           }
                        }
                        /********** END 设置币种对应的计算进程 ************/


                        /********** 设置币种对应的监控 ************/
                        if(!array_key_exists($symbol,$symbolWebsocketList)) {
                            $tmp = array_count_values($symbolWebsocketList);
                            $sortList = $symbolWebsocketCount;
                            foreach($tmp as $key=>$val) {
                                $sortList[$key] = $val;
                            }
                            asort($sortList);
                            foreach($sortList as $key=>$del) {
                                $symbolWebsocketList[$symbol] = $key;
                                break;
                            }
                        }
                        /******** END 设置币种对应的监控 **********/
                    }
                    $tmpList = $symbolIniList;
                    foreach($symbolWebsocketList as $symbol=>$value) {
                        if(!in_array($symbol,$symbolList)) {
                            unset($symbolWebsocketList[$symbol]);
                            continue;
                        }
                        $tmpList[$value][] = $symbol;
                    }
                    foreach($tmpList as $index=>$symbol) {
                        $WebsocketToSymbolTable->set($index,['symbol_list'=>implode(',',$symbol)]);
                    }
                }catch (\Exception $e) {
                }
                usleep($this->usleepTimeDefault * 5);
            }
        });



        return true;

        Swoole::startProcess(function() use($calculationProcessList){
            // 添加监控进程
            Swoole::monitorProcess('行情管理');

            $symbolTable = Swoole::table(static::TABLE_SELECT_LIST);
            $symbolRunTable = Swoole::table(static::TABLE_SELECT_RUN);
            $symbolToPindex = [];
            $pindexList = [];
            foreach($calculationProcessList as $key=>$del) {
                $pindexList[$key] = 0;
            }
            unset($del);
            while (true) {

                foreach ($symbolTable as $symbol=>$value) {
                    if(!($runDetails = $symbolRunTable->get($symbol)) || time() - 10 > $runDetails['lastScessTime']) {
                        if(!array_key_exists($symbol,$symbolToPindex)) {
                            // 设置进程
                            asort($pindexList);
                            foreach($pindexList as $index=>$count) {
                                $pindexList[$index]++;
                                $symbolToPindex[$symbol] = $index;
                                break;
                            }
                        }
                        $process = $calculationProcessList[$symbolToPindex[$symbol]];
                        // 开启等待线程，并执行处理
                        Swoole::startWaitProcess(function($symbol) use($process){
                            return $this->getTickers($symbol,$process);
                        },$symbol,'getTickersMaster');



                    }
                }
                usleep($this->usleepTimeDefault * 3);
                Swoole::closeWaitBySub();
            }
        },true,true);
    }


    /**
     * 获取用户的key列表 及 货币被监控的用户列表
     */
    private function getUserKeysAndStrategys()
    {
        $strategys = [];
        $result = self::getUserRule(self::USER_RULE_CONFIG['TYPE_LIST']['HISTORY']);
       // $result = (new SqlToRedis())->getUserRule();
        foreach($result as $ruleDetails) {
            if(!array_key_exists($ruleDetails['code'],$strategys)) {
                $strategys[$ruleDetails['code']] = [];
            }
            $strategys[$ruleDetails['code']][] = $ruleDetails['id'];
        }
        foreach($strategys as $key=>$value) {
            $strategys[$key] = [
                'ruleIdList' => implode('|',$value)
            ];
        }

        // 返回列表数据
        return [$strategys];
    }



    /**
     * 启动tcp输出表数据进程
     */
    private function createTcpQuery() {
        Swoole::startProcess(function(){
            $host = '0.0.0.0';
            $port = static::tcpQueryPort();
            $serv = new \swoole_server($host,$port);

            //swoole处理方法
            $serv->set([
                'worker_num'    => self::TCP_WORKER_NUM,     // worker process num
                'open_eof_check' => true,   //打开EOF检测
                'package_eof'    => "\r\n", //设置EOF
            ]);


            $serv->on('connect',function($serv, $fd){
            });
            $serv->on('receive',function($serv, $fd, $reactor_id, $data){
                $data = substr($data,0,-2);
                $send = [
                    'status'    => 'error',
                    'time'  => microtime(true),
                    'param' => $data,
                    'result'  => null
                ];
                try {
                    $send['status'] = 'ok';
                    $data = explode(' ',trim($data));
                    switch($data[0]) {
                        case 'setInput':
                            $params = array_slice($data,1);
                            $send['result'] = $this->setInput($params);
                            break;
                        case 'userRuleRefresh':   // 刷新用户规则
                            $send['result'] = static::userRuleRefresh();
                            break;
                        case 'selectSymbol':    // 获取所有监控的交易对
                            if($table = Swoole::tableGet(self::TABLE_SELECT_LIST)) {
                                $symbolList = [];
                                foreach($table as $symbol=>$del) {
                                    $symbolList[] = $symbol;
                                }
                                $send['result'] = $symbolList;
                            }
                            break;
                        case 'apiReply':    // 接收api请求的回复
                            $table = Swoole::tableGet(RuleCalculation::TABLE_LAST_REQUEST);
                            $table->lockFunc(function($table) use($data){
                                if(!$record = $table->get($data[1])) {
                                    return false;
                                }
                                ksort($record);
                                if(md5(json_encode($record)) == $data[2]) {
                                    $record = array_merge($record,[
                                       'status' => $data[3],
                                       'last_run_time'  => time()
                                    ]);
                                    $table->set($data[1],$record);
                                }
                            });
                            break;
                        case 'takeTime':    // 添加 taketime 日志
                            HistoryTrade::takeTime($data[1],$data[2],$data[3],$data[4],$data[5],$data[6]);
                            break;
                        default:
                            if($table = Swoole::tableGet($data[0])) {

                                if(array_key_exists(1,$data)) {
                                    if(($send['result'] = $table->get($data[1])) && array_key_exists(2,$data)) {
                                        $send['result'] = array_key_exists($data[2],$send['result'])?$send['result'][$data[2]]:null;
                                    }
                                } else {
                                    $send['result'] = [];
                                    foreach($table as $key=>$value) {
                                        $send['result'][$key] = $value;
                                    }
                                }

                            }
                    }

                } catch (\Exception $e) {
                    $send['status'] = 'error';
                    $send['result'] = $e->getMessage();
                }


                $serv->send($fd, json_encode($send));
            });
            $serv->on('close',function($serv, $fd){
            });
            $serv->start();
        });
    }

    /**
     * 设置输入价格
     * @param $params
     */
    private function setInput($params) {
        if(!$table = Swoole::tableGet(static::TABLE_INPUT_PRICE)) {
            return false;
        }
        list($symbol,$price) = $params;
        $table->set('market_' . $symbol,['price'=>$price]);
        return $table->lockFunc(function($table) use($symbol,$price) {
            $priceList = '';
            if($details = $table->get($symbol)) {
                $priceList = $details['price'];
            }
            $priceList = explode(',',$priceList);
            $priceList = array_filter($priceList);
            $priceList[] = $price;
            $table->set($symbol,['price'=>implode(',',$priceList)]);
            return true;
        });
    }

    /**
     * 获取输入的火币价格 【用于测试】
     * @param $symbol
     */
    private function getInput($symbol) {
            $result = [
                'ch'    => 'market.xlmusdt.trade.detail',
                'status'    => 'ok',
                'ts'    => time() * 1000,
                'data'  => []
            ];
            if(!$table = Swoole::tableGet(static::TABLE_INPUT_PRICE)) {
                return $result;
            }

            if($list = $table->lockFunc(function($table) use($symbol) {
                if(!$details = $table->get($symbol)) {
                    return [];
                }
                $table->del($symbol);
                return explode(',',$details['price']);
            })) {
                $i = 0;
                foreach($list as $price) {
                    if(!array_key_exists($i,$result['data'])) {
                        $result['data'][$i] = [
                            'id'    => rand(1000000000,1099999999),
                            'ts'    => ceil( bcmul(microtime(true),1000,0)),
                            'data'  => [],
                        ];
                    }
                    $result['data'][$i]['data'][] = [
                        'id'    => rand(100000000,1099999999),
                        'ts'    => $result['data'][$i]['ts'],
                        'trade-id'  => rand(100000000,1099999999),
                        'amount'    => rand(10,9999),
                        'price' => $price,
                        'direction' => 'sell'
                    ];
                    $i += rand(0,1);
                }
            }
            return $result;
    }


    /**
     * 获取用户的配置信息
     * @param string $type
     */
    static function getUserRule(string $type) {
        $table = Swoole::tableGet(static::USER_RULE_CONFIG['TABLE_NAME']);
        if(($config = $table->get($type)) && time() - $config['last_time'] < static::USER_RULE_CONFIG['INTERVAL_TIME']) {
            throw new \Exception('No update required');
        }
        // 设置最新查询时间
        $table->set($type,[
            'last_time' => time(),
        ]);
        $result = (new SqlToRedis())->getUserRule(null,true,true);
      //  echo date('Y-m-d H:i:s',time()),"\n";
        try {
            Db::close();
        }catch (\Exception $e) {
        }

        return $result;
    }

    /**
     * 刷新用户规则
     */
    static function userRuleRefresh() {
        try {
            $table = Swoole::tableGet(static::USER_RULE_CONFIG['TABLE_NAME']);
            foreach(static::USER_RULE_CONFIG['TYPE_LIST'] as $type) {
                $table->set($type,[
                    'last_time' => 0,
                ]);
            }
        }catch (\Exception $e) {
            return false;
        }
        return true;
    }


}